from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, IntegerType, StringType, LongType
from pyspark.sql.functions import col, from_unixtime

# 创建 SparkSession
spark = SparkSession.builder \
    .appName("NezhaCommentAnalysis") \
    .getOrCreate()

# 手动定义 Schema（确保列名正确）
schema = StructType([
    StructField("index", IntegerType(), True),
    StructField("用户id", StringType(), True),
    StructField("性别", StringType(), True),
    StructField("评论时间", LongType(), True),
    StructField("点赞数", IntegerType(), True),
    StructField("评论", StringType(), True),
    StructField("回复量", IntegerType(), True)
])

# 读取 CSV 文件（指定编码和 Schema）
df = spark.read.csv(
    "哪吒去重.csv",
    header=True,
    schema=schema,
    encoding="GBK",  # 根据实际文件编码调整
    escape='"'
)

# # 清洗数据（去重、删除空值）
# df_cleaned = df.dropDuplicates().na.drop()

# # 转换时间戳
# df_transformed = df_cleaned.withColumn(
#     "评论日期",
#     from_unixtime(col("评论时间")).cast("timestamp")
# ).drop("评论时间")
#
# # 查看结果
# df_transformed.printSchema()
# df_transformed.show(5, truncate=False)

# 根据所有字段去重
df_cleaned = df.dropDuplicates()

# 或者根据特定字段去重（例如用户ID和评论时间）
# df_cleaned = df.dropDuplicates(["用户id", "评论时间"])

# 删除包含空值的行（根据需求选择）
df_cleaned = df_cleaned.na.drop()

# 或者填充缺失值（例如填充性别为“未知”）
# df_cleaned = df_cleaned.na.fill({"性别": "未知"})

df_transformed = df_cleaned.withColumn(
    "评论日期",
    from_unixtime(col("评论时间")).cast("timestamp")
).drop("评论时间")

# 查看转换后的时间
df_transformed.select("评论日期").show(5, truncate=False)

from pyspark.sql.functions import udf
from pyspark.sql.types import IntegerType

# 定义UDF统计关键词
def count_keyword(comment, keyword):
    return comment.count(keyword) if comment else 0

count_keyword_udf = udf(lambda x: count_keyword(x, "哪吒"), IntegerType())

df_transformed = df_transformed.withColumn(
    "哪吒关键词出现次数",
    count_keyword_udf(col("评论"))
)

# 查看结果
df_transformed.select("评论", "哪吒关键词出现次数").show(5, truncate=False)

gender_stats = df_transformed.groupBy("性别") \
    .count() \
    .withColumnRenamed("count", "评论数量") \
    .orderBy("评论数量", ascending=False)

gender_stats.show()

from pyspark.sql.functions import avg

avg_stats = df_transformed.agg(
    avg("点赞数").alias("平均点赞数"),
    avg("回复量").alias("平均回复量")
)

avg_stats.show()

top_comments = df_transformed.orderBy(col("回复量").desc()) \
    .select("评论", "回复量") \
    .limit(10)

top_comments.show(truncate=False)

df_transformed.write.mode("overwrite") \
    .csv("cleaned_nezhacomments", header=True)

print("=== 性别统计 ===")
gender_stats.show()

print("=== 平均指标 ===")
avg_stats.show()

print("=== 热门评论 ===")
top_comments.show(truncate=False)

# from pyspark.sql import SparkSession
# from pyspark.sql.types import StructType, StructField, IntegerType, StringType, LongType
# from pyspark.sql.functions import col, from_unixtime
#
# # 创建 SparkSession
# spark = SparkSession.builder \
#     .appName("NezhaCommentAnalysis") \
#     .getOrCreate()
#
# # 手动定义 Schema（确保列名正确）
# schema = StructType([
#     StructField("index", IntegerType(), True),
#     StructField("用户id", StringType(), True),
#     StructField("性别", StringType(), True),
#     StructField("评论时间", LongType(), True),
#     StructField("点赞数", IntegerType(), True),
#     StructField("评论", StringType(), True),
#     StructField("回复量", IntegerType(), True)
# ])
#
# # 读取 CSV 文件（指定编码和 Schema）
# df = spark.read.csv(
#     "哪吒去重.csv",
#     header=True,
#     schema=schema,
#     encoding="GBK",  # 根据实际文件编码调整
#     escape='"'
# )
#
# # 清洗数据（去重、删除空值）
# df_cleaned = df.dropDuplicates().na.drop()
#
# # 转换时间戳
# df_transformed = df_cleaned.withColumn(
#     "评论日期",
#     from_unixtime(col("评论时间")).cast("timestamp")
# ).drop("评论时间")
#
# # 查看结果
# df_transformed.printSchema()
# df_transformed.show(5, truncate=False)